package com.abyss.watermark;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Arrays;

/**
 * 侧输出
 */
public class SideOutputDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        DataStreamSource<String> source = env.socketTextStream("node1", 9999);

        SingleOutputStreamOperator<Tuple2<String, Long>> map = source.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                return Tuple2.of(value.split(" ")[0], Long.parseLong(value.split(" ")[1]));
            }
        });

        // watermarks
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = map.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(3)) {
            @Override
            public long extractTimestamp(Tuple2<String, Long> element) {
                return element.f1;
            }
        });

        // allowedLateness
        WindowedStream<Tuple2<String, Long>, Tuple, TimeWindow> allowedLateness = watermarks.keyBy(0).timeWindow(Time.seconds(5)).allowedLateness(Time.seconds(2));

        // side output   OutputTag对象必须是匿名内部类的形式创建出来, 本质上得到的是OutputTag对象的一个匿名子类
        OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("side output"){};
        WindowedStream<Tuple2<String, Long>, Tuple, TimeWindow> sideOutputLateData =
                allowedLateness.sideOutputLateData(outputTag);

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = sideOutputLateData.apply(new WindowFunction<Tuple2<String, Long>, Tuple2<String, Integer>, Tuple, TimeWindow>() {
            @Override
            public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<Tuple2<String, Integer>> out) throws Exception {
                String key = null;
                int counter = 0;
                for (Tuple2<String, Long> ele : input) {
                    key = ele.f0;
                    counter++;
                }
                out.collect(Tuple2.of(key, counter));
            }
        });

        result.print();

        DataStream<Tuple2<String, Long>> sideOutput = result.getSideOutput(outputTag);
        // 对得到的保存超级迟到数据的DataStream进行处理
        sideOutput.print("late>>>");
        env.execute();
    }
}